package com.atguigu.realtime.app

import com.atguigu.realtime.bean.AdsInfo
import com.atguigu.realtime.util.{MyApp, RedisUtil}
import org.apache.spark.streaming.Minutes
import org.apache.spark.streaming.dstream.DStream
import org.json4s.jackson.JsonMethods
import redis.clients.jedis.Jedis

/**
 * description ：最近 1 小时广告点击量实时统计（每分钟点击量）
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/1/19 17:24
 * modified By ：
 * version:    : 1.0
 */
object LastHourAdsClickApp extends MyApp {

  override def doSomething(adsInfo: DStream[AdsInfo]): Unit = {

    // 先设置好窗口参数,及转换数据结构 ((adsid,17:38),1),((adsid,17:39),1)
    val adsIdAndhmStringOne: DStream[((String, String), Int)] = adsInfo.window(Minutes(60)).map(ai => ((ai.adsId, ai.hmString), 1))

    // ((adsid,17:38),1),((adsid,17:39),1) reduceByKey -->  ((adsid,17:38),count),((adsid,17:39),count)
    val adsIdAndhmStringCount: DStream[((String, String), Int)] = adsIdAndhmStringOne.reduceByKey(_ + _)

    // ((adsid,17:38),count),((adsid,17:39),count) map --> adsid, (17:38,count1),(17:39,count1) --> groupbyKey
    val result: DStream[(String, String)] = adsIdAndhmStringCount.map {
      case ((adsid, hmString), count) => (adsid, (hmString, count))
    }.groupByKey
      .mapValues { // 这里将value的值转换为了string,方便后面直接转换为map,写入reids
        it =>
          import org.json4s.JsonDSL._
          JsonMethods.compact(JsonMethods.render(it))
      }

    // 对结果写到redis  key --> last:hour:ads:click filed --> 广告id  value --> ((17:38,count1),(17:39,count1))
    result.foreachRDD {
      rdd =>
        val reduClient: Jedis = RedisUtil.getJedisClient
        import collection.JavaConversions._
        println("------------" + rdd.collect.toMap)
        reduClient.hmset("ast:hour:ads:click", rdd.collect.toMap)
        reduClient.close()
    }
  }
}

/*
最近 1 小时广告点击量实时统计（每分钟点击量）
((adsid,17:38),1),((adsid,17:39),1)
((adsid,17:38),count),((adsid,17:39),count)
adsid, (17:38,count1),(17:39,count1)
reids
key --> last:hour:ads:click filed --> 广告id  value --> ((17:38,count1),(17:39,count1))
 */
